package com.atcyj.gamll.realtime.common.util;

import com.atcyj.gamll.realtime.common.constant.Constant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * @author: cyj
 * @date: 2024/8/14 
 */
public class FlinkSourceUtil {
    public static KafkaSource<String> getKafkaSource(String groupId, String topic) {

        return KafkaSource.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setTopics(topic)
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(
                        // 如果有null值，用new SimpleStringSchema()会报错
                        // new SimpleStringSchema();
                        // 实现一个反序列化接口，模仿SimpleStringSchema就好
                        new DeserializationSchema<String>() {
                            @Override
                            public String deserialize(byte[] message) throws IOException {
                                if (message != null && message.length != 0) {
                                    return new String(message, StandardCharsets.UTF_8);
                                }
                                return "";
                            }

                            @Override
                            public boolean isEndOfStream(String nextElement) {
                                return false;
                            }

                            @Override
                            public TypeInformation<String> getProducedType() {
                                return BasicTypeInfo.STRING_TYPE_INFO;
                            }
                        }
                )
                .build();
    }

    public static MySqlSource<String> getMySqlSource(String database, String table) {
        // mysql 8.0可能会出现不允许通过密码连接的情况
        // useSSL=false
        Properties props = new Properties();
        props.setProperty("useSSL", "false");
        props.setProperty("allowPublicKeyRetrieval", "true");
        return MySqlSource.<String>builder()
                .jdbcProperties(props)
                .hostname(Constant.MYSQL_HOST)
                .port(Constant.MYSQL_PORT)
                .username(Constant.MYSQL_USER_NAME)
                .password(Constant.MYSQL_PASSWORD)
                .databaseList(database)
                .tableList(database + "." + table)
                .deserializer(new JsonDebeziumDeserializationSchema()) // json格式
                .startupOptions(StartupOptions.initial()) // 初始化读取：先把整张表数据读一下，后续有变更就抓取
                .build();
    }
}
